package com.snda.dw.lidejia.hdfsoverftp;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HdfsOverFtpSystem {

    private static final Logger LOGGER = LoggerFactory.getLogger(HdfsOverFtpSystem.class);

    private static FileSystem HDFS = null;
    public static String HDFS_URI = "";
    private static String SUPERUSER = "superuser";
    private static String SUPERGROUP = "supergroup";

    private static void hdfsInit() {
        Path path = new Path(HDFS_URI);

        Configuration conf = new Configuration();
        conf.set("hadoop.job.ugi", SUPERUSER + "," + SUPERGROUP);
        try {
            FileSystem fs = path.getFileSystem(conf);
//            HDFS.set(fs);
            HDFS = fs;
            LOGGER.info(String.format("HDFS home directory is %s", fs));
        } catch (Exception e) {
            LOGGER.error("HDFS Initialization error", e);
            throw new RuntimeException("DFS Initialization error");
        }
    }

//    private static void hdfsInit(User user) {
//        //Path p = new Path(HDFS_URI);
//
//        Configuration conf = new Configuration();
//        //conf.set("hadoop.job.ugi", user.getName() + "," + supergroup);
//
//        try {
//            //dfs = p.getFileSystem(conf);
//            HDFS = FileSystem.get(URI.create("hdfs://name101.dev.la.mezimedia.com/mezi/quicky/dev/smus/public"), conf);
//            LOGGER.info("HDFS home directory is " + HDFS.getHomeDirectory());
//            LOGGER.info("HDFS working directory is " + HDFS.getWorkingDirectory());
//        } catch (Exception e) {
//            LOGGER.error("DFS Initialization error", e);
//            throw new RuntimeException("DFS Initialization error");
//        }
//    }

    /*public static void setHDFS_URI(String HDFS_URI) {
    HDFS_URI = HDFS_URI;
    }*/
    public static void setHDFS_URI(String hdfs_uri) {
        HDFS_URI = hdfs_uri;
    }

    public static synchronized  FileSystem getDfs() {
//        FileSystem fs = HDFS.get();
        FileSystem fs = HDFS;
        if (fs == null) {
            hdfsInit();
        }
//        return HDFS.get();
        return HDFS;
    }

//    public static FileSystem getDfs(User user) {
//        if (HDFS == null) {
//            hdfsInit(user);
//        }
//        return HDFS;
//    }
    public static void setSuperuser(String superuser) {
        SUPERUSER = superuser;
    }
}